package bigdata.hourglass.engine.core

import java.util.concurrent.atomic.AtomicReference
import java.util.{Map => JMap}

import bigdata.hourglass.common.Logging
import org.apache.spark.sql.SparkSession
import org.apache.spark.{LSQLConf, SparkConf}

import scala.collection.JavaConversions._

/**
 * @author leo.jie (weixiao.me@aliyun.com)
 * @organization DataReal
 * @version 1.0
 * @website https://www.jlpyyf.com
 * @date 2019/12/11 10:30 下午
 * @since 1.0
 */
class SparkRuntime(_params: JMap[Any, Any]) extends Logging {
  val params: JMap[Any, Any] = _params
  val configReader = LSQLConf.createConfigReader(params.map(f => (f._1.toString, f._2.toString)))

  def createRuntime(sparkConf: SparkConf): SparkSession = {
    logger.info("create spark session ......")
    params.filter(f => f._1.toString.startsWith("spark.") ||
      f._1.toString.startsWith("hive."))
      .foreach {
        f => sparkConf.set(f._1.toString, f._2.toString)
      }

    def isLocalMaster(master: String): Boolean = {
      master.startsWith("local[")
    }

    if (LSQLConf.L_SQL_MASTER.readFrom(configReader).isDefined &&
      isLocalMaster(LSQLConf.L_SQL_MASTER.readFrom(configReader).get)) {
      sparkConf.setMaster(LSQLConf.L_SQL_MASTER.readFrom(configReader).get)
    }

    sparkConf.setAppName(LSQLConf.L_SQL_NAME.readFrom(configReader))

    val sparkSession = SparkSession.builder().config(conf = sparkConf)

    if (LSQLConf.L_SQL_ENABLE_HIVE_SUPPORT.readFrom(configReader)) {
      sparkSession.enableHiveSupport()
    }
    sparkSession.getOrCreate()
  }
}

object SparkRuntime {
  private val INSTANTIATION_LOCK = new Object()
  @transient private val lastInstantiatedContext = new AtomicReference[SparkRuntime]()

  /*def getOrCreate(params: JMap[Any, Any]): SparkRuntime = {
    INSTANTIATION_LOCK.synchronized {
      if (lastInstantiatedContext.get() == null) {
        new SparkRuntime(params)
      }
    }
    lastInstantiatedContext.get()
  }*/

  def getOrCreate(params: JMap[Any, Any]): SparkRuntime = {
    new SparkRuntime(params)
  }
}